package com.cet.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

public class FlowDemo {

    public static void main(String[] args) throws InterruptedException {

        // 发布者,推送数据到缓存区
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        // 订阅者
        Flow.Subscriber<String> subscriber1 = new Flow.Subscriber<>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread().getName() + ": 订阅开始了，" + subscription);
                this.subscription = subscription;

                // 请求一个数据
                subscription.request(1);
            }


            @Override
            public void onNext(String item) {
                System.out.println(Thread.currentThread().getName() + ": 订阅者接收到数据：" + item);
                if ("p-7".equals(item)) {
                    subscription.cancel();
                } else {
                    subscription.request(1);
                }
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread().getName() + ": 订阅者接收到错误信号，" + throwable);
            }

            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread().getName() + ": 管道处理完成");
            }
        };

        Flow.Subscriber<String> subscriber2 = new Flow.Subscriber<>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println(Thread.currentThread().getName() + ": 订阅开始了，" + subscription);
                this.subscription = subscription;

                // 请求一个数据
                subscription.request(1);
            }


            @Override
            public void onNext(String item) {
                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.println(Thread.currentThread().getName() + ": 订阅者接收到数据：" + item);
                if ("p-7".equals(item)) {
                    subscription.cancel();
                } else {
                    subscription.request(1);
                }
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println(Thread.currentThread().getName() + ": 订阅者接收到错误信号，" + throwable);
            }

            @Override
            public void onComplete() {
                System.out.println(Thread.currentThread().getName() + ": 管道处理完成");
            }
        };

        // 绑定发布者和订阅者
        publisher.subscribe(subscriber1);
        publisher.subscribe(subscriber2);

        // 订阅关系绑定完成后，发布者推送数据，订阅者才能接受到
        for (int i = 0; i < 10; i++) {
            if (i == 5) {
                TimeUnit.SECONDS.sleep(1);
                publisher.closeExceptionally(new RuntimeException("p-5"));
                break;
            } else {
                publisher.submit("p-" + i);
            }
        }
        publisher.close();

        TimeUnit.SECONDS.sleep(5);
    }
}
